package com.silas.dynflightdatatransfer.sharding.algorithm;

import cn.hutool.core.date.DatePattern;
import cn.hutool.core.date.DateUtil;
import cn.hutool.extra.spring.SpringUtil;
import com.google.common.collect.Range;
import com.silas.dynflightdatatransfer.common.constants.Constants;
import com.silas.dynflightdatatransfer.manager.FlightDynamicDataInfoManager;
import org.apache.commons.lang3.StringUtils;
import org.apache.shardingsphere.sharding.api.sharding.standard.PreciseShardingValue;
import org.apache.shardingsphere.sharding.api.sharding.standard.RangeShardingValue;
import org.apache.shardingsphere.sharding.api.sharding.standard.StandardShardingAlgorithm;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

/**
 * sharding分表规则：根据上报时间戳进行按月分割
 *
 * @author chentianyan
 */
@Order(value = 1)
@Component
public class TimeStampSplitByMonth implements StandardShardingAlgorithm<Long> {

    private static final Logger LOGGER = LoggerFactory.getLogger(TimeStampSplitByMonth.class);

    public static final Set<String> TABLE_NAME_SET = ConcurrentHashMap.newKeySet();

    /**
     * 时间戳 转 年月yyyyMM
     *
     * @param timeStamp 时间戳
     * @return 年月yyyyMM
     */
    private String convertYearMonth(Long timeStamp) {
        // todo 需要改成按月 SIMPLE_MONTH_FORMATTER, 方便调试改成按天 PURE_DATE_FORMATTER
        return DateUtil.date(timeStamp).toLocalDateTime().format(DatePattern.SIMPLE_MONTH_FORMATTER);
    }

    /**
     * 查询时返回对应分表名 <br/>
     * 注：查询条件必须带有上报时间戳time_stamp列，否则不会触发
     */
    @Override
    public Collection<String> doSharding(Collection<String> collection, RangeShardingValue<Long> rangeShardingValue) {
        // 查询缓存Set中的表
        Set<String> tableNameCache = TABLE_NAME_SET;
        // 获取查询条件 精确匹配表
        Range<Long> valueRange = rangeShardingValue.getValueRange();
        // 开始时间戳
        Long beginLong = valueRange.lowerEndpoint();
        // 结束时间戳
        Long endLong = valueRange.upperEndpoint();
        // 开始时间-yyyyMM
        String begin = convertYearMonth(beginLong);
        // 结束时间-yyyyMM
        String end = convertYearMonth(endLong);

        List<String> queryTables = new ArrayList<>(tableNameCache.size());
        tableNameCache.forEach(t -> {
            // 不匹配逻辑表名
            if (!StringUtils.equals(rangeShardingValue.getLogicTableName(), t)) {
                // 截取分表名后缀的年月-yyyyMM
                String yearMon = t.substring(t.lastIndexOf(Constants.UNDER_LINE) + 1);
                // 在查询时间范围内
                if (Integer.decode(yearMon) >= Integer.decode(begin)
                        && Integer.decode(yearMon) <= Integer.decode(end)) {
                    queryTables.add(t);
                }
            }
        });
        // 返回按条件匹配的表名（如没有匹配的表名，组件会处理默认去查询逻辑表）
        return queryTables;
    }

    /**
     * 写入数据时，返回需要写入的分表名 <br/>
     * 注：如表不存在，则需要先创建分表
     */
    @Override
    public String doSharding(Collection<String> collection, PreciseShardingValue<Long> preciseShardingValue) {
        StringBuilder tableName = new StringBuilder();
        // 逻辑表名
        String logicTableName = preciseShardingValue.getLogicTableName();
        // 拼接分表名称：逻辑表名_yyyyMM
        tableName.append(logicTableName)
                .append(Constants.UNDER_LINE)
                .append(convertYearMonth(preciseShardingValue.getValue()));
        // 缓存Set中已存在，则直接返回
        if (TABLE_NAME_SET.contains(tableName.toString())) {
            return tableName.toString();
        }
        // 创建新分表
        SpringUtil.getBean(FlightDynamicDataInfoManager.class).createTable(tableName.toString(), logicTableName);
        // 加入缓存Set中
        TABLE_NAME_SET.add(tableName.toString());
        // 返回写入的分表名
        return tableName.toString();
    }

    @Override
    public void init() {
 System.out.println("111");
    }

    @Override
    public String getType() {
        //　自定义
        return null;
    }
}
